package com.jugheadzhou.kafka.consumer.listener;

import com.fasterxml.jackson.core.type.TypeReference;
import com.jugheadzhou.kafka.consumer.entity.PatientBase;
import com.jugheadzhou.kafka.consumer.service.PatientBaseService;
import com.jugheadzhou.kafka.core.base.KafkaMessage;
import com.jugheadzhou.kafka.core.enums.KafkaMessageTypeEnum;
import com.jugheadzhou.kafka.core.utils.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * @author jugheadzhou
 * @date 2021-09-23
 */
@Slf4j
@Service
public class TestTopicListener {

    private final static String TOPIC = "test-topic-auto";

    @Resource
    private PatientBaseService patientBaseService;

    @KafkaListener(topics = {TOPIC})
    public void consumerForTestTopic(ConsumerRecord<String, String> consumerRecord) {
        log.info("topic={} partition={} key={}, value={}",
                consumerRecord.topic(),
                consumerRecord.partition(),
                consumerRecord.key(),
                consumerRecord.value()
        );
        KafkaMessage<PatientBase> kafkaMessage = JsonUtil.string2Object(
                consumerRecord.value(),
                new TypeReference<KafkaMessage<PatientBase>>() {
                }
        );
        if (kafkaMessage == null) {
            log.error("消息接收失败：{}", consumerRecord);
            return;
        }
        log.info("kafkaMessage.toString={}", kafkaMessage);
        // TODO 模拟消息按类型分发
        if (KafkaMessageTypeEnum.MRH_SAVE.getCode().equals(kafkaMessage.getType())) {
            patientBaseService.savePatientBase(kafkaMessage.getData());
        } else if (KafkaMessageTypeEnum.MRH_PRINT.getCode().equals(kafkaMessage.getType())) {
            patientBaseService.printPatientBase(kafkaMessage.getData());
        }
    }

}
